package com.yihaodian.cluster;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;

import com.google.common.collect.Maps;
import com.yihaodian.index.meta.IndexMeta;
import com.yihaodian.index.meta.IndexMetas;
import com.yihaodian.index.meta.routing.RoutingTable;
import com.yihaodian.index.meta.routing.RoutingTables;
import com.yihaodian.index.meta.shard.Shard;
import com.yihaodian.index.meta.shard.ShardId;
import com.yihaodian.index.meta.shard.ShardRouting;

public class NewClusterStateBuilders {

	static Map<ClusterStateUpdateType, NewClusterStateBuilder> builders = new HashMap<ClusterStateUpdateType, NewClusterStateBuilder>();

	static {
		builders.put(ClusterStateUpdateType.cluster_state_init,
				new InitStateBuilder());
		builders.put(ClusterStateUpdateType.cluster_node_change,
				new NodeChangeBuilder());
		builders.put(ClusterStateUpdateType.cluster_routing_table_change,
				new RoutingTableBuilder());
		builders.put(ClusterStateUpdateType.cluster_index_meta_change,
				new IndexMetaBuilder());
		builders.put(ClusterStateUpdateType.shard_data_version_change,
				new IndexVersionBuilder());
		builders.put(ClusterStateUpdateType.shard_mgt_change,
				new ShardMgtBuilder());
	}

	public static NewClusterStateBuilder getBuilder(ClusterStateUpdateType type) {
		return builders.get(type);
	}

	static class InitStateBuilder implements NewClusterStateBuilder {
		private static Logger logger = Logger.getLogger(InitStateBuilder.class);

		@Override
		public ClusterState newClusterState(ClusterStateUpdate update,
				ClusterState state) {
			ClusterState zkClusterState = update.getState();
			ClusterState localState = state;
			Node masterNode = update.getMasterNode();

			Cluster cluster = new Cluster.Builder()
					.setId(masterNode.getClusterId()).addNode(masterNode)
					.setMaster(masterNode).build();

			ClusterState newClusterState = null;
			if (localState != null
					&& localState.version() > zkClusterState.version) {
				logger.info("local cluster state has new version "
						+ localState.version);

				newClusterState = new ClusterState.Builder(localState).cluster(
						cluster).build();
			} else {
				newClusterState = new ClusterState.Builder(zkClusterState)
						.cluster(cluster).build();
			}

			newClusterState = reset(newClusterState);

			return newClusterState;
		}

		/**
		 * reset running status.
		 * 
		 * @param cs
		 * @return
		 */
		ClusterState reset(ClusterState cs) {
			RoutingTables rts = cs.routingTables();
			for (String index : rts.indexNames()) {
				RoutingTable rt = rts.routingTable(index);
				rt.resetStatus();
			}

			return cs;
		}

	}

	static class NodeChangeBuilder implements NewClusterStateBuilder {
		@Override
		public ClusterState newClusterState(ClusterStateUpdate update,
				ClusterState state) {
			Collection<Node> nodeMap = update.getNodeList();
			Map<String, Node> ids = Maps.newHashMap();
			for (Node node : nodeMap)
				ids.put(node.getId(), node);

			Cluster.Builder clusterBuilder = new Cluster.Builder(state.cluster);
			Collection<Node> nodes = state.cluster.getNodes();

			// remove
			for (Node node : nodes) {
				if (!ids.containsKey(node.getId()))
					clusterBuilder.removeNode(node);
			}

			// add
			for (String id : ids.keySet()) {
				if (!clusterBuilder.contains(id)) {
					clusterBuilder.addNode(ids.get(id));
				}
			}

			/**
			 * TODO RoutingTables change IndexMetas change
			 */
			return new ClusterState.Builder(state).cluster(
					clusterBuilder.build()).build();
		}
	}

	static class RoutingTableBuilder implements NewClusterStateBuilder {

		@Override
		public ClusterState newClusterState(ClusterStateUpdate update,
				ClusterState state) {
			RoutingTable rt = update.getRoutingTable();
			RoutingTables rts = new RoutingTables.Builder(state.routingTables)
					.add(rt).build();
			IndexMeta meta = state.indexMetas().indexMeta(rt.getIndexId())
					.clone();
			if (meta == null || meta.getShards() == null
					|| meta.getShards().isEmpty()) {
				return new ClusterState.Builder(state).routingTables(rts)
						.build();
			} else {
				Collection<Shard> shards = meta.getShards();
				for (Shard shard : shards) {
					Collection<String> nodes = rt.getNodes(shard.getShardId());
					if (nodes == null || nodes.size() == 0) {
						shard.setEnable(false);
					} else {
						shard.setEnable(true);
					}
				}
				IndexMetas metas = new IndexMetas.Builder(state.indexMetas())
						.add(meta).build();
				return new ClusterState.Builder(state).routingTables(rts)
						.indexMetas(metas).build();
			}

		}
	}

	static class IndexMetaBuilder implements NewClusterStateBuilder {
		@Override
		public ClusterState newClusterState(ClusterStateUpdate update,
				ClusterState state) {
			IndexMeta meta = update.getIndexMeta();

			IndexMetas metas = new IndexMetas.Builder(state.indexMetas()).add(
					meta).build();
			return new ClusterState.Builder(state).indexMetas(metas).build();
		}
	}

	static class IndexVersionBuilder implements NewClusterStateBuilder {
		@Override
		public ClusterState newClusterState(ClusterStateUpdate update,
				ClusterState state) {
			Map<ShardId, String> dataVersion = update.getDataVersion();

			IndexMetas.Builder builder = new IndexMetas.Builder(
					state.indexMetas());
			for (Entry<ShardId, String> entry : dataVersion.entrySet()) {
				ShardId shardId = entry.getKey();
				String version = entry.getValue();

				IndexMeta meta = state.indexMetas().indexMeta(
						shardId.getIndexId()).clone();
				meta.setVersion(shardId.getShardId(), version);

				builder.add(meta);
			}

			return new ClusterState.Builder(state).indexMetas(builder.build())
					.build();
		}
	}

	static class ShardMgtBuilder implements NewClusterStateBuilder {

		@Override
		public ClusterState newClusterState(ClusterStateUpdate update,
				ClusterState state) {
			ShardMgtChange change = update.getShardMgtChange();
			ShardId id = change.getShardId();
			boolean isEnable = change.isEnable();

			List<Node> nodeList = new ArrayList<Node>();
			IndexMetas metas = state.indexMetas();
			if (change.isAllNode()) {
				IndexMeta meta = state.indexMetas().indexMeta(id.getIndexId());

				IndexMeta newMeta = meta.clone();
				newMeta.getShard(id).setEnable(isEnable);
				metas = new IndexMetas.Builder(state.indexMetas()).add(newMeta)
						.build();
				nodeList.addAll(state.cluster.getNodes());

			} else {
				nodeList.add(change.getNode());
			}

			RoutingTable rt = state.routingTables.routingTable(id.getIndexId());
			RoutingTable newRt = rt.clone();

			for (Node node : nodeList) {
				Collection<ShardRouting> srs = newRt.getShardRouting(node
						.getId());
				for (ShardRouting sr : srs) {
					if (sr.getShardId().equals(id)) {
						sr.setEnable(isEnable);
						break;
					}
				}
			}
			RoutingTables rts = new RoutingTables.Builder(state.routingTables)
					.add(newRt).build();

			return new ClusterState.Builder(state).indexMetas(metas)
					.routingTables(rts).build();
		}
	}

}
